package cn.doitedu.rtmk.demo6;

import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/24
 * @Desc: 学大数据，上多易教育
 * 实现规则的动态上下线、以及规则静态画像人群的预圈选功能
 **/

@Slf4j
public class Demo6 {

    //public static final  Logger logger = LoggerFactory.getLogger(Demo5.class);

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);




        /* *
         *  一、 接收用户行为事件流
         */
        // 创建kafka 行为明细topic的source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doit43-8")
                .setClientIdPrefix("doitedu-c")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");
        SingleOutputStreamOperator<UserEvent> eventStream = streamSource.map(json -> JSON.parseObject(json, UserEvent.class));

        KeyedStream<UserEvent, Long> keyedEventStream = eventStream.keyBy(UserEvent::getUser_id);


        /* *
         *  二、接收规则元数据表的cdc元数据流
         */

        MySqlSource<String> cdcSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("doit43")
                .tableList("doit43.rule_meta")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> metaCdcStream = env.fromSource(cdcSource, WatermarkStrategy.noWatermarks(), "meta_cdc");

        // 解析cdc json，封装成我们自己的RuleMeta对象
        SingleOutputStreamOperator<RuleMeta> ruleMetaStream = metaCdcStream.map(new MapFunction<String, RuleMeta>() {
            @Override
            public RuleMeta map(String cdcJson) throws Exception {

                JSONObject jsonObject = JSON.parseObject(cdcJson);
                String op = jsonObject.getString("op");

                RuleMeta ruleMeta;
                if (op.equals("d")) {
                    ruleMeta = jsonObject.getObject("before", RuleMeta.class);
                } else {
                    ruleMeta = jsonObject.getObject("after", RuleMeta.class);
                }

                // 填充op字段
                ruleMeta.setOp(op);

                // 反序列化预圈选人群的bitmap的字节数组
                byte[] rulePreSelectUsersBytes = ruleMeta.getRule_pre_select_users();
                Roaring64Bitmap preSelectUsersBitmap = Roaring64Bitmap.bitmapOf();

                if (rulePreSelectUsersBytes != null) {
                    preSelectUsersBitmap.deserialize(ByteBuffer.wrap(rulePreSelectUsersBytes));
                }
                // 将反序列化好的bitmap填充到ruleMeta对象
                ruleMeta.setPreSelectUsersBitmap(preSelectUsersBitmap);


                return ruleMeta;
            }
        });

        // 将规则元数据流广播
        MapStateDescriptor<String, String> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, String.class);
        BroadcastStream<RuleMeta> ruleMetaBroadcastStream = ruleMetaStream.broadcast(bcStateDesc);


        /* *
         *  三、将行为事件keyed流  连接  规则元数据广播流
         */
        keyedEventStream.connect(ruleMetaBroadcastStream)
                .process(new KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, String>() {

                    HashMap<String, RuleCalculator> ruleCalculatorPool;
                    GroovyClassLoader groovyClassLoader;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ruleCalculatorPool = new HashMap<>();
                        groovyClassLoader = new GroovyClassLoader();
                    }

                    // 处理用户行为事件数据（keyed流中的数据）
                    @Override
                    public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {

                        // 遍历运算机池
                        for (Map.Entry<String, RuleCalculator> entry : ruleCalculatorPool.entrySet()) {
                            RuleCalculator ruleCalculator = entry.getValue();
                            // 调用规则运算机对象，处理当前收到的用户行为事件
                            ruleCalculator.calculate(userEvent, out);
                        }

                    }

                    // 处理规则元数据（broadcast流中的数据）
                    @Override
                    public void processBroadcastElement(RuleMeta ruleMeta, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMeta, String>.Context ctx, Collector<String> out) throws Exception {

                        String ruleId = ruleMeta.getRule_id();
                        String modelId = ruleMeta.getModel_id();
                        String op = ruleMeta.getOp();

                        if (op.equals("d")) {
                            // 根据ruleId，从规则运算机池中移除该规则的运算机对象
                            ruleCalculatorPool.remove(ruleId);
                            log.warn("下线了一个规则，规则id:{}", ruleId);

                        } else {

                            // 用groovy的类加载工具，在运行时，根据规则附带的运算机模型代码，构造运算机实例
                            Class aClass = groovyClassLoader.parseClass(ruleMeta.getModel_calculator_source_code());
                            RuleCalculator ruleCalculator = (RuleCalculator) aClass.newInstance();

                            // 从规则元数据中，提取各种必须的信息：规则参数、规则人群、runtimeContext
                            String ruleParamJson = ruleMeta.getRule_param_json();
                            Roaring64Bitmap preSelectUsersBitmap = ruleMeta.getPreSelectUsersBitmap();
                            RuntimeContext runtimeContext = getRuntimeContext();

                            // 利用元数据中的信息，初始化这个新的规则实例对象
                            ruleCalculator.init(runtimeContext, ruleParamJson, preSelectUsersBitmap);

                            // 将初始化好的规则运算机对象，放入全局运算机池
                            ruleCalculatorPool.put(ruleId, ruleCalculator);

                            log.warn("上线了一个规则，规则id:{},规则人群:{}", ruleId, preSelectUsersBitmap);
                        }


                    }
                }).print();


        env.execute();
    }
}
